package org.rcy.framework.mq.producer;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.rcy.framework.api.mq.AbstractMessage;
import org.rcy.framework.mq.queue.KafkaDelayQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.concurrent.TimeUnit;

/**
 * @author 吴康桥
 * @Description
 * @date 2022/11/23 10:42
 */

public class KafkaMessageProducer<M extends AbstractMessage> implements MessageProducer<M>{

	private final Logger logger = LoggerFactory.getLogger(this.getClass());
	private KafkaTemplate<Object,Object> kafkaTemplate;
	private KafkaDelayQueue delayQueue;

	public KafkaMessageProducer() {
	}

	public KafkaMessageProducer(KafkaTemplate<Object, Object> kafkaTemplate, KafkaDelayQueue delayQueue) {
		this.kafkaTemplate = kafkaTemplate;
		this.delayQueue = delayQueue;
	}

	public KafkaMessageProducer(KafkaTemplate<Object, Object> kafkaTemplate) {
		this.kafkaTemplate = kafkaTemplate;
	}

	@Override
	public void produce(M message) {
		try {
			String topic = message.getTopic();
			this.kafkaTemplate.send(new ProducerRecord<>(topic,message));
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 *
	 * @param message
	 * @param delay
	 * @param timeUnit 单位只支持秒和分钟
	 */
	@Override
	public void produce(M message, Long delay, TimeUnit timeUnit) {
		try {
			String topic = message.getTopic();
			switch (timeUnit){
				case SECONDS: delay = delay * 1000; break;
				case MINUTES: delay = delay * 60000; break;
				default: break;
			}
			delayQueue.add(topic,message,delay);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
}
